package cn.dmp.tags

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.Row

object Tags4KeyWords extends Tags{
    /**
      * 行数据 -》 标签数据
      *
      * @param args
      * @return
      */
    override def makeTags(args: Any*): Map[String, Int] = {

        val row = args(0).asInstanceOf[Row]
        val stopwordsBroadcast = args(1).asInstanceOf[Broadcast[Map[String, Int]]]

        var tags = Map[String, Int]()

        // 关键字
        val keywords = row.getAs[String]("keywords")
        keywords.split("\\|")
          .filter(k => k.length >=3
            && k.length <= 8
            && !stopwordsBroadcast.value.contains(k)).foreach(k => tags += "K"+k -> 1)

        tags
    }
}
